package com.rabbit.demo.ylr.direct;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

//import sun.nio.cs.UTF_8;

/**
 * @ClassName Consumer
 * @Description TODO
 * @Author 11
 * @Date 2021/7/14 11:07
 * @Version 1.0
 */
public class Consumer {
    private static Runnable runnable = () -> {
//        1. 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
//        2.设置连接属性
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = null;
        Channel channel = null;

        String name = Thread.currentThread().getName();


        try {
//        3. 从连接属性中获取连接
            connection = factory.newConnection("消费者");
//        4. 从连接中获取通道channel
            channel = connection.createChannel();

//        申明要消费的队列
            channel.queueDeclare(name, true, false, false, null);

//        绑定队列到交换机

            channel.queueBind(name, "direct-exchange", "testkey1");

//        basicQos
            channel.basicQos(1);

//        回调的消费者处理类

            Channel finalChannel = channel;
            com.rabbitmq.client.Consumer consumer = new DefaultConsumer(finalChannel) {

                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    // 接收到的消息
                    String message = new String(body);
                    System.out.println(" [1] Received '" + message + "'");
                }
            };

//        消费消息
            channel.basicConsume(name, true, consumer);


            System.out.println(name + " start accept ！");

        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } finally {
            if (channel != null && channel.isOpen())
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
        }


    };

    /**
    * 主函数
    * <br/>
     * @param args	
    * @return void
    * @author ylr
    * @date 2021/7/19 10:45
    **/
    public static void main(String[] args) {
        new Thread(runnable, "q1").start();
        new Thread(runnable, "q2").start();
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
//        runnable.run();
    }
}
